read_serial_sensor

读取串口传感器数据

接触了很多串口通信的传感器，基本都殊途同归，就是发送一串数据，

其中包括固定的数据头，真实数据，校验和以及固定的数据尾。

用keil写过C版本的，用arduino写过c++版本的，用树莓派写过python版本的，这里记录一下。

读取的思路就是编写一个函数，这个函数每次就读取一个字节，循环调用这个函数。

函数内部就做一些标志位的判断，判断是在等待数据头还是接受数据又或者是判断校验和。

当一组数据解析完成之后再将数据更新。

以我最近使用的空气质量传感器为例，

C版本的是之前的IMU，就不放出来了，道理和C++是一样的，用结构体实现。

商品详情是这样子写的

c++代码

#include "AirQualitySensor.h"

AirQualitySensor air_quality_sensor;
void AirQualitySensor::init(HardwareSerial* serial_ptr)
{   
    serial_ptr->begin(9600);
    this->serial_ptr = serial_ptr; //传入串口指针 有需要可以很方便修改成其他串口
}

const short data_length = 14;// 数据长度共14位
const short buf_bength = 17;// buf长度17位
unsigned char data_buf[buf_bength]={0};//包括数据头、数据、校验和 共17位

void AirQualitySensor::update_data()
{
    static unsigned char flag = Waiting;//状态标志位 
    static unsigned char ReceiverFront = 0;//上一次接受的数据
    static short CurrentReceiverDataNum = 0;//目前接受了多少个数据

    unsigned char Receiver = 0;//当前接受的数据
    if(serial_ptr->available()>0)
    {
        Receiver =serial_ptr->read();
    }

    //等待
    if(flag == Waiting)
    {
        //数据头
        if((ReceiverFront == 0x3C) && (Receiver == 0x02))
        {
            flag = Started;
            data_buf[0] = ReceiverFront;
            data_buf[1] = Receiver;         
        }
        else
        {
            ReceiverFront = Receiver;
        }
    }
    //获取数据
    else
    {
        //将数据放入buf 包括校验和
        if(CurrentReceiverDataNum < data_length +1)
        {
            data_buf[CurrentReceiverDataNum+2] = Receiver;
            CurrentReceiverDataNum ++;
        }
        //数据读取完毕 检验校验和 通过则对buf解析放入类的属性
        else
        {
            unsigned char CheckSum = 0;
            for(int i=0;i<2+data_length;i++)
            {
                CheckSum += data_buf[i];
            }

            if ((CheckSum&0xff) == data_buf[buf_bength-1])
            {
                air_quality_sensor.CO2 = (data_buf[2]<<8) + data_buf[3];
                air_quality_sensor.CH2O = (data_buf[4]<<8) + data_buf[5];
                air_quality_sensor.TVOC = (data_buf[6]<<8) + data_buf[7];
                air_quality_sensor.PM2_5 = (data_buf[8]<<8) + data_buf[9];
                air_quality_sensor.PM10 = (data_buf[10]<<8) + data_buf[11];
                if(((data_buf[12] & 0x0040)>>6)==0)
                {
                    air_quality_sensor.temperature = CombineIntegerDecimal(data_buf[12],data_buf[13]);
                }
                else
                {
                    air_quality_sensor.temperature = -CombineIntegerDecimal(data_buf[12],data_buf[13]);
                }

                
                air_quality_sensor.humidity = CombineIntegerDecimal(data_buf[14],data_buf[15]);             
            }
            else
            {
                Serial.println("CheckSum error");
            }

            //不论校验和通不通过都重置参数
            ReceiverFront =0;
            CurrentReceiverDataNum =0;
            flag = Waiting;

        }
    }   
}

//合并整数和小数
float CombineIntegerDecimal(unsigned char integer,unsigned char decimal)
{
	// 将整数和小数拼接成字符串再转回数字
	String combine = String(integer) +"."+String(decimal);
	float result = combine.toFloat();

    return result;
}

#pragma once
#include <Arduino.h>
class  AirQualitySensor
{
    private:
        HardwareSerial* serial_ptr;
 
    public:
        int CO2;//二氧化碳
        int CH2O;//甲醛
        int TVOC;//总挥发性有机物
        int PM2_5;//PM2.5
        int PM10;//PM10
        float temperature;//温度
        float humidity;//湿度

        void init(HardwareSerial* serial_ptr);
        void update_data();

};

float CombineIntegerDecimal(unsigned char integer,unsigned char decimal);
enum GetDataFlag
{
    Waiting, //等待
    Started //开始
};

python代码

# coding=utf-8
from enum import Enum

class Flag(Enum):
    Waiting = 0
    Started = 1

def bytes2int(byte):
    return int.from_bytes(byte,"little",signed=False) #python读串口最麻烦的就是编码问题 注意这里要用python3

class AirQualitySensor:

    def __init__(self, serial):
        self._serial = serial
        
        self.data_length = 14  # 有效数据长度
        self.buf_length = 17  # buf长度
        self.data_buf = [0] * self.buf_length  # 包括帧头、帧尾和校验和

        self.receiver = 0  # 当前接受到的数据
        self.receiver_front = 0  # 上一帧接受到的数据
        self.now_data_num = 0  # 目前接受了多少个数据
        self.flag = Flag.Waiting  # 当前flag 用于判断帧头是否满足 进而判断是否接受数据

        # 数据
        self.CO2 = 0
        self.CH2O = 0
        self.TVOC = 0
        self.PM2_5 = 0
        self.PM10 = 0
        self.temperature = 0
        self.humidity = 0

    def update_data(self):      
        self.receiver = bytes2int(self._serial.read(1)) # 接受数据

        # 等待数据头校验通过
        if self.flag == Flag.Waiting:
            if self.receiver_front == bytes2int(b'\x3C') and \
               self.receiver == bytes2int(b'\x02'):             
                self.data_buf[0] = self.receiver_front
                self.data_buf[1] = self.receiver
                
                self.flag = Flag.Started
            else:
                self.receiver_front = self.receiver

        # 获取数据
        else:
            # 将数据放入buf 包括最后一位校验和
            if self.now_data_num + 2 < self.buf_length:
                self.data_buf[self.now_data_num + 2] = self.receiver
                self.now_data_num += 1
                
            # 数据读取完毕 校验校验和
            else:
                check_sum = sum(self.data_buf[:-1])                              
                if (check_sum & 0x00ff) == self.data_buf[-1]:
                    # 将数据放入成员变量
                    self.CO2 = (self.data_buf[2] << 8) + self.data_buf[3]
                    self.CH2O = (self.data_buf[4] << 8) + self.data_buf[5]
                    self.TVOC = (self.data_buf[6] << 8) + self.data_buf[7]
                    self.PM2_5 = (self.data_buf[8] << 8) + self.data_buf[9]
                    self.PM10 = (self.data_buf[10] << 8) + self.data_buf[11]
                    self.humidity = float(str(self.data_buf[14]) + "." + str(self.data_buf[15]))
                    # 温度需要判断正负
                    if ((self.data_buf[12] & 0x0040) >> 6) == 0:
                        self.temperature = float(str(self.data_buf[12]) + "." + str(self.data_buf[13]))
                    else:
                        self.temperature = -float(str(self.data_buf[12]) + "." + str(self.data_buf[13]))
                else:
                    print("check sum error!")

                # 不论是否通过检验 都重置
                self.receiver = 0
                self.now_data_num = 0
                self.flag = Flag.Waiting